package com.suxg.hbase.test;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;

import com.suxg.hbase.api.RowMapper;
import com.suxg.hbase.model.ClassInfoModel;
import com.suxg.hbase.model.F;
import com.suxg.hbase.model.SchoolModel;
import com.suxg.hbase.model.StudentModel;
import com.suxg.hbase.model.TeacherModel;
import com.suxg.hbase.utils.HBaseUtilsFilters2;
import com.suxg.hbase.utils.HBaseUtilsMapper;

public class HBaseUtilsTest7 {

	private static final String TABLE_NAME = "n_test_columnMuits";
	private static final String SCHOOL = "school";
	private static final String CLASS_INFO = "classInfo";
	private static final String TEACHER = "teacher";
	private static final String STUDENT = "student";

	private static Connection connection;

	static {
		// localhost
		Configuration configuration = HBaseConfiguration.create();
		configuration.set("hbase.zookeeper.property.clientPort", "2181");
		// 如果是集群 则主机名用逗号分隔
		configuration.set("hbase.zookeeper.quorum", "localhost:2181");
		configuration.set("hbase.nodeParent", "localhost:2181");
		// aliyun
//		Configuration configuration = HBaseConfiguration.create();
//		configuration.set("hbase.zookeeper.property.clientPort", "2181");
//		// 如果是集群 则主机名用逗号分隔
//		configuration.set("hbase.zookeeper.quorum", "101.200.212.148:2181");
//		configuration.set("hbase.nodeParent", "101.200.212.148:2181");

		try {
			connection = ConnectionFactory.createConnection(configuration);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public static void main(String[] args) throws Exception {
		long startTime = System.currentTimeMillis();
		Table table = connection.getTable(TableName.valueOf(TABLE_NAME));
//		创建1000行数据
//		createAndInsert(1000000);
//		按rowkey 按rokey,列族取出当前rowkye列族下所有字段值
//		SchoolModel t=(SchoolModel) getRow(TABLE_NAME,"rowKey999",TEACHER,new SchoolModel());
//		System.out.println(t.getXx1()+t.getMoney());
//		按rokey、列族、列 ，为条件取出当列字段值
//		String cellValue=getFamliyCell(TABLE_NAME, "rowKey999", SCHOOL, "firstName");
//		System.out.println("cell firstName :" + cellValue);

//		//获取单个列族下的数据
//		Filter filter1 = new FamilyFilter(CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(SCHOOL)));
//		需要注意的是在多台 Regin Services 上执行分页过滤的时候，由于并行执行的过滤器不能共享它们的状态和边界，所以有可能每个过滤器都会在完成扫描前获取了 PageCount 行的结果，这种情况下会返回比分页条数更多的数据，分页过滤器就有失效的可能。
//		Filter filter2 = new PageFilter(20); //分页过滤,当数据超过50w可数据不准确
//		FilterList filterList = new FilterList(filter2);
//		//
//		List<SchoolModel> list = getScannerList(table, "rowKey980", "rowKey990", filterList, SchoolModel.class);
//
//		long endTime = System.currentTimeMillis();
//		System.out.println((endTime - startTime) / 1000.0);
//		System.out.println("共查出记录" + list.size());
//		
//		
//		for(SchoolModel sm:list) {
//			System.out.println("----"+sm.getRowKey()+ "--shool="+sm.getXx8()+sm.getXx5()
//				 );
//		}

//		for(TestColumnMuitsModel sm:list) {
//			System.out.println("----"+sm.getRowKey()+ "--shool="+sm.getSchoolModel().getXx9()
//					+"----class="+sm.getClassInfoModel().getXx9()+"----teacher="+sm.getTeacherModel().getName());
//		}
//		
		String tableName = "playback_gateway_log";
		List<Put> putList = new ArrayList<>();
		byte [] family="F".getBytes();
		byte [] qualifier="uuid".getBytes();
		
		byte [] value="3428328-23424-23423-42-34".getBytes();
		ByteBuffer bb=ByteBuffer.wrap(value);
		Put p=new Put(bb).addImmutable(family, qualifier, value);
		putList.add(p);
	
		boolean b = HBaseUtilsFilters2.putBath(tableName, putList);
		System.out.println("#############" + b);
	}

	/**
	 * 创建N条数据
	 * 
	 * @param num
	 */
	public static void createAndInsert(int num) {
		try {
			if (HBaseUtilsFilters2.tableExists(TABLE_NAME)) {
// 		删除表
				deleteTable();
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		// 新建表
		List list = new ArrayList();
		list.add(SCHOOL);
		list.add(CLASS_INFO);
		list.add(TEACHER);
		list.add(STUDENT);
		createTable(TABLE_NAME, list);

		// 插入表
		String[] key = new String[num];

		for (int i = 0; i < num; i++) {

			key[i] = "rowKey" + i;
			String[] columnFamilyNames = { SCHOOL, CLASS_INFO, TEACHER, STUDENT };
			for (int j = 0; j < columnFamilyNames.length; j++) {
				String columnFamilyName = columnFamilyNames[j];
				List<Pair<String, String>> pairList = new ArrayList<>();
				switch (columnFamilyName) {
				case SCHOOL:
					buildDataSchool(pairList, i);
					break;
				case CLASS_INFO:
					buildData(pairList, CLASS_INFO);
					break;
				case TEACHER:
					Pair<String, String> p1 = new Pair<String, String>("name", "xx1");
					Pair<String, String> p2 = new Pair<String, String>("gender", "xx2");
					Pair<String, String> p3 = new Pair<String, String>("money", "100.01");
					Pair<String, String> p4 = new Pair<String, String>("age", "18");
					Pair<String, String> p5 = new Pair<String, String>("firstName", "xx5");
					Pair<String, String> p6 = new Pair<String, String>("lastName", "xx6");
					Pair<String, String> p7 = new Pair<String, String>("position", "xx7");
					Pair<String, String> p8 = new Pair<String, String>("salary", "xx8");
					Pair<String, String> p9 = new Pair<String, String>("aaa", "xx9");
					Pair<String, String> p10 = new Pair<String, String>("list", "xx10");
					pairList.add(p1);
					pairList.add(p2);
					pairList.add(p3);
					pairList.add(p4);
					pairList.add(p5);
					pairList.add(p6);
					pairList.add(p7);
					pairList.add(p8);
					pairList.add(p9);
					pairList.add(p10);
					break;
				case STUDENT:
					Pair<String, String> p11 = new Pair<String, String>("name", "xx1student");
					Pair<String, String> p21 = new Pair<String, String>("age", "18");
					Pair<String, String> p31 = new Pair<String, String>("gender", "100");
					pairList.add(p11);
					pairList.add(p21);
					pairList.add(p31);
					break;
//         		  ...

				}

				insertData(TABLE_NAME, key[i], columnFamilyNames[j], pairList);

			}

		}

//    	insertData();
//		getRow();
//		getScanner("oldp_ns:test2", "20_20210422 ", "F");
//		getClassScanner("class", "rowKey1", "student",new StudentMapper(),new StudentModel() );

//		getFScanner("oldp_ns:test2", "20_20210422", "F",new FMapper());
//    	getCell();
//		getScanner();
//		getScannerWithFilter();
//		// 获取单个RowKey 所有数据
//		StudentModel s = (StudentModel) HBaseUtilsFilters.getRow(TABLE_NAME, ROWKEY, STUDENT, new StudentModel());
//		// 获取指定Rowkey, 列族指定列的值
//		String signValue = HBaseUtilsFilters.getCell(TABLE_NAME, ROWKEY, STUDENT, "name");
//		// 扫描全表
//		List<StudentModel> list = HBaseUtilsFilters.ScannerList(TABLE_NAME, StudentModel.class);
//
//		// JAVA 8 流排序
//		List<StudentModel> students = list.stream().
//
//				sorted((Comparator.comparingInt(StudentModel::getAge).reversed()))
//
//				.collect(Collectors.toList());
//		// 单个字段升序排序
//		List<StudentModel> signList = (List<StudentModel>) SortListUtils.sort(list, "age");
//
//		// 多字段排序 fields 里的值为要排序的字段, sorts 里为排序的方法与fields里的对应，为null默认生序
//		String[] fields = { "gender"  };
//		String[] sorts = { SortListUtils.DESC};
//		@SuppressWarnings("unchecked")
//		List<StudentModel> sortList = (List<StudentModel>) SortListUtils.sort(list, fields, sorts);

		// 性能测试

//		
//		Table table = getConnection().getTable(TableName.valueOf(TABLE_NAME));
//		Scan scan = new Scan();
//		scan.setStartRow("rowKey1".getBytes());
//		scan.setStopRow("rowKey2000".getBytes());
//		
//		scan.setMaxResultsPerColumnFamily(100);
//		long startTime=System.currentTimeMillis();
//		List<StudentModel> scannerList = HBaseUtilsFilters.ScannerList(scan, table, StudentModel.class);
//		long endTime=System.currentTimeMillis();
//		System.out.println("所使用时间"+(endTime-startTime)/1000.0);
//		System.out.println();
//		
//		System.out.println(s.getAge() + "###获取单个RowKey###" + s.getName());
//
//		System.out.println("#####获取单个值###" + signValue);
//
//		System.out.println(list.size());
//		for (StudentModel sm : list) {
//			System.out.println(sm.getName() + "------" + sm.getAge() + "------" + sm.getGender());
//		}
//
//		for (StudentModel sm : students) {
//			System.out.println("###JAVA 8 流排序 测试#"+sm.getName() + "------" + sm.getAge() + "------" + sm.getGender());
//		}
//		
//		for (StudentModel sm : signList) {
//			System.out.println("###单个字段升序排序 测试#"+sm.getName() + "------" + sm.getAge() + "------" + sm.getGender());
//		}
//		
//		for (StudentModel sm : sortList) {
//			System.out.println("###sortList排序测试#"+sm.getName() + "------" + sm.getAge() + "------" + sm.getGender());
//		}

//		for (StudentModel sm : scannerList) {
//			System.out.println("###性能测试#"+sm.getName() + "------" + sm.getAge() + "------" + sm.getGender());
//		}
	}

	/**
	 * 构建模拟数据
	 * 
	 * @param pairList
	 */
	public static void buildData(List<Pair<String, String>> pairList, String key) {
		Pair<String, String> p1 = new Pair<String, String>("xx1", "xx1" + key);
		Pair<String, String> p2 = new Pair<String, String>("xx2", "xx2" + key);
		Pair<String, String> p3 = new Pair<String, String>("xx3", "xx3" + key);
		Pair<String, String> p4 = new Pair<String, String>("xx4", "xx4" + key);
		Pair<String, String> p5 = new Pair<String, String>("xx5", "xx5" + key);
		Pair<String, String> p6 = new Pair<String, String>("xx6", "xx6" + key);
		Pair<String, String> p7 = new Pair<String, String>("xx7", "xx7" + key);
		Pair<String, String> p8 = new Pair<String, String>("xx8", "xx8" + key);
		Pair<String, String> p9 = new Pair<String, String>("xx9", "xx9" + key);
		Pair<String, String> p10 = new Pair<String, String>("xx10", "xx10" + key);
		pairList.add(p1);
		pairList.add(p2);
		pairList.add(p3);
		pairList.add(p4);
		pairList.add(p5);
		pairList.add(p6);
		pairList.add(p7);
		pairList.add(p8);
		pairList.add(p9);
		pairList.add(p10);
	}

	public static void buildDataSchool(List<Pair<String, String>> pairList, int key) {
		Pair<String, String> p1 = new Pair<String, String>("xx1", "xx1School" + key);
		Pair<String, String> p2 = new Pair<String, String>("xx2", "xx2School" + key);
		Pair<String, String> p3 = new Pair<String, String>("xx3", "xx3School" + key);
		Pair<String, String> p4 = new Pair<String, String>("xx4", "xx4School" + key);
		Pair<String, String> p5 = new Pair<String, String>("xx5", "xx5School" + key);
		Pair<String, String> p6 = new Pair<String, String>("xx6", "xx6School" + key);
		Pair<String, String> p7 = new Pair<String, String>("xx7", "xx7School" + key);
		Pair<String, String> p8 = new Pair<String, String>("xx8", "xx8School" + key);
		Pair<String, String> p9 = new Pair<String, String>("xx9", "xx9School" + key);
		Pair<String, String> p10 = new Pair<String, String>("xx10", "xx10School" + key);
		pairList.add(p1);
		pairList.add(p2);
		pairList.add(p3);
		pairList.add(p4);
		pairList.add(p5);
		pairList.add(p6);
		pairList.add(p7);
		pairList.add(p8);
		pairList.add(p9);
		pairList.add(p10);
	}

	/**
	 * 创建表
	 */
	public static void createTable(String tableName, List<String> columnFamilies) {
		boolean table = HBaseUtilsFilters2.createTable(TABLE_NAME, columnFamilies);
		if (table) {
			System.out.println("创建表成功");
		} else {
			System.out.println("创建表失败");
		}
	}

	/**
	 * 插入表
	 */
	public static void insertData(String tableName, String key, String columnFamilyName,
			List<Pair<String, String>> pairList) {

		HBaseUtilsFilters2.putRow(tableName, key, columnFamilyName, pairList);
	}

	/**
	 * 按rokey、列族，取出当前rowkye列族下所有字段值
	 * 
	 * @param tableName
	 * @param rowKey
	 * @param famliy
	 * @param object
	 * @return
	 */
	public static Object getRow(String tableName, String rowKey, String famliy, Object object) {
		Object result = null;
		try {
			result = HBaseUtilsFilters2.getRow(tableName, rowKey, famliy, object);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		return result;

	}

	// 按rokey、列族、列 ，为条件取出当列字段值
	public static String getFamliyCell(String tableName, String rowKey, String famliy, String cell) {
		String cellValue = HBaseUtilsFilters2.getCell(tableName, rowKey, famliy, cell);
		return cellValue;
	}

	// 全表查询
	public static <T> List<T> getScannerList(Table tableName, String startRowKey, String endRowKey,
			FilterList filterList, Class<T> c) {
		ResultScanner resultScanner = HBaseUtilsFilters2.getScanner(tableName, startRowKey, endRowKey, filterList,
				SCHOOL.getBytes());

		List<T> list = HBaseUtilsFilters2.convertScannerToList(resultScanner, c);

		Map<String, Class> map = new HashMap<>();
		map.put(SCHOOL, SchoolModel.class);
		map.put(CLASS_INFO, ClassInfoModel.class);
		map.put(TEACHER, TeacherModel.class);
		map.put(STUDENT, StudentModel.class);
//		ResultScanner resultScanner = HBaseUtilsFilters2.getScanner(tableName, startRowKey, endRowKey, filterList,null);
//		List<T> list =  (List<T>) HBaseUtilsFilters2.convertScannerCellToListMuilts(resultScanner,map, c);

		return list;

	}

	/**
	 * 对Hbase的ResultScanner结果中单个列族做转换
	 * 
	 * @param <T>
	 * @param resultScanner
	 * @param c
	 * @return
	 */
	public static <T> List<T> convertScannerToList(ResultScanner resultScanner, Class<T> c) {

		// 上面部分还按原来方法写，该分页的分页，该过滤的过滤，我只要resultScanner的结果做转换
		List<T> list = HBaseUtilsFilters2.convertScannerToList(resultScanner, c);
		return list;
//		for (StudentModel sm : list) {
//			System.out.println(sm.getName() + sm.getAge() + sm.getGender());
//		}

	}

	/**
	 * 对Hbase的ResultScanner结果中多个列族做转换
	 * 
	 * @param <T>
	 * @param resultScanner
	 * @param c
	 * @return
	 */
	public static <T> List<T> convertScannerToListMuilts(ResultScanner resultScanner, Map<String, Class> map1,
			Class<T> tableClass) {

		Map<String, Class> map = new HashMap<>();
		map.put(SCHOOL, SchoolModel.class);
		map.put(CLASS_INFO, ClassInfoModel.class);
		map.put(TEACHER, TeacherModel.class);
		map.put(STUDENT, StudentModel.class);

		List<T> list = (List<T>) HBaseUtilsFilters2.convertScannerToListMuilts(resultScanner, map, tableClass);
		return list;
//		for (StudentModel sm : list) {
//			System.out.println(sm.getName() + sm.getAge() + sm.getGender());
//		}

	}

	public static void getFScanner(String tableName, String rowKey, String fimeys, RowMapper rowMappers, Object o) {
		List<F> list = HBaseUtilsMapper.getHbaseObjectList(tableName, rowMappers, rowKey, fimeys, o);
		for (F f : list) {
			System.out.println(f.getCid() + "###f###" + f.getSdf());
		}

	}

	public static void getClassScanner(String tableName, String rowKey, String fimeys, RowMapper rowMappers, Object o) {
		List<StudentModel> list = HBaseUtilsMapper.getHbaseObjectList(tableName, rowMappers, rowKey, fimeys, o);
		System.out.println(list.size());
		for (StudentModel s : list) {
			System.out.println(s.getAge() + "###u###" + s.getName());
		}

	}

	public static void getScannerWithFilter() {
		FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
		SingleColumnValueFilter nameFilter = new SingleColumnValueFilter(Bytes.toBytes(STUDENT), Bytes.toBytes("name"),
				CompareOp.EQUAL, Bytes.toBytes("Jack"));
		filterList.addFilter(nameFilter);
		ResultScanner scanner = HBaseUtilsMapper.getScanner(TABLE_NAME, filterList);
		if (scanner != null) {
			scanner.forEach(result -> System.out.println(Bytes.toString(result.getRow()) + "->"
					+ Bytes.toString(result.getValue(Bytes.toBytes(STUDENT), Bytes.toBytes("name")))));
			scanner.close();
		}
	}

	public static void deleteColumn() {
		boolean b = HBaseUtilsMapper.deleteColumn(TABLE_NAME, "rowKey2", STUDENT, "age");
		System.out.println("删除结果: " + b);
	}

	public static void deleteRow() {
		boolean b = HBaseUtilsMapper.deleteRow(TABLE_NAME, "rowKey2");
		System.out.println("删除结果: " + b);
	}

	public static void deleteTable() {
		boolean b = HBaseUtilsMapper.deleteTable(TABLE_NAME);
		System.out.println("删除结果: " + b);
	}

	public static Connection getConnection() {
		Connection connection = null;
		// localhost
		Configuration configuration = HBaseConfiguration.create();
		configuration.set("hbase.zookeeper.property.clientPort", "2181");
		// 如果是集群 则主机名用逗号分隔
		configuration.set("hbase.zookeeper.quorum", "localhost:2181");
		configuration.set("hbase.nodeParent", "localhost:2181");
		// aliyun
//	Configuration configuration = HBaseConfiguration.create();
//	configuration.set("hbase.zookeeper.property.clientPort", "2181");
//	// 如果是集群 则主机名用逗号分隔
//	configuration.set("hbase.zookeeper.quorum", "101.200.212.148:2181");
//	configuration.set("hbase.nodeParent", "101.200.212.148:2181");

		try {
			connection = ConnectionFactory.createConnection(configuration);

		} catch (IOException e) {
			e.printStackTrace();
		}

		return connection;
	}

}
